package com.atguigu.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.TableProcess;
import com.atguigu.common.Constant;
import com.atguigu.util.HBaseUtil;
import com.atguigu.util.JdbcUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.sql.Connection;
import java.sql.DriverManager;
import java.util.*;

public class DimDwdTableProcessFunction extends BroadcastProcessFunction<JSONObject, TableProcess, JSONObject> {

    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;
    private OutputTag<JSONObject> outputTag;
    private HashMap<String, TableProcess> tableProcessHashMap;

    public DimDwdTableProcessFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptor, OutputTag<JSONObject> outputTag) {
        this.mapStateDescriptor = mapStateDescriptor;
        this.outputTag = outputTag;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //初始化HashMap
        tableProcessHashMap = new HashMap<>();

        //预加载配置信息  从MySQL查询一次
        Connection mysqlConn = DriverManager.getConnection(Constant.MYSQL_URL, "root", "000000");
        org.apache.hadoop.hbase.client.Connection hbaseConn = HBaseUtil.getConnection();
        List<TableProcess> queryList = JdbcUtil.queryList(mysqlConn, "select * from `gmall-230524-config`.table_process", TableProcess.class, true);
        for (TableProcess tableProcess : queryList) {
            if ("dim".equals(tableProcess.getSinkType())) {
                byte[][] splitKeys = HBaseUtil.getSplitKeys(tableProcess.getSinkExtend());
                HBaseUtil.createTable(hbaseConn, Constant.HBASE_NAME_SPACE, tableProcess.getSinkTable(), splitKeys, tableProcess.getSinkFamily());
                tableProcessHashMap.put(tableProcess.getSourceTable(), tableProcess);
            } else {
                //dwd
                tableProcessHashMap.put(tableProcess.getSourceTable() + "-" + tableProcess.getSourceType(), tableProcess);
            }
        }

        //关闭连接
        mysqlConn.close();
        hbaseConn.close();
    }

    @Override
    public void processElement(JSONObject value, BroadcastProcessFunction<JSONObject, TableProcess, JSONObject>.ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {

        String dimKey = value.getString("table");
        String type = value.getString("type");
        String dwdKey = dimKey + "-" + type;

        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);

        ///////////////DIM/////////////////////
        TableProcess dimTableProcess = broadcastState.get(dimKey);
        TableProcess dimTableProcessMap = tableProcessHashMap.get(dimKey);
        if (dimTableProcess == null) {
            dimTableProcess = dimTableProcessMap;
        }
        if (dimTableProcess != null && !"bootstrap-start".equals(type) && !"bootstrap-complete".equals(type)) {
            //列过滤
            filterColumns(value.getJSONObject("data"), dimTableProcess.getSinkColumns());

            //插入未来需要的数据
            value.put("sink_table", dimTableProcess.getSinkTable());
            value.put("row_key_column", dimTableProcess.getSinkRowKey());
            value.put("sink_extend", dimTableProcess.getSinkExtend());
            value.put("sink_column_family", dimTableProcess.getSinkFamily());

            //写出
            out.collect(value);
        } else {
            System.out.println(dimKey + "不是维表或者操作类型不匹配:" + type);
        }

        ///////////////DWD/////////////////////
        TableProcess dwdTableProcess = broadcastState.get(dwdKey);
        TableProcess dwdTableProcessMap = tableProcessHashMap.get(dwdKey);
        if (dwdTableProcess == null) {
            dwdTableProcess = dwdTableProcessMap;
        }
        if (dwdTableProcess != null) { //属于DWD数据
            //列过滤
            filterColumns(value.getJSONObject("data"), dwdTableProcess.getSinkColumns());
            //补充信息
            value.put("sink_topic", dwdTableProcess.getSinkTable());
            //写出
            ctx.output(outputTag, value);
        } else {
            System.out.println(dimKey + "不属于DWD数据或者操作类型不匹配:" + type);
        }
    }

    @Override
    public void processBroadcastElement(TableProcess value, BroadcastProcessFunction<JSONObject, TableProcess, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {

        //获取广播状态数据
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);

        //获取sink_type
        String sinkType = value.getSinkType();
        if ("dim".equals(sinkType)) {
            //获取op字段
            String op = value.getOp();
            //获取source_table字段
            String sourceTable = value.getSourceTable();
            //如果sink_type为dim,Op为"d"   则删除状态中的数据
            if ("d".equals(op)) {
                broadcastState.remove(sourceTable);
                tableProcessHashMap.remove(sourceTable);
            } else {
                //如果sink_type为dim,Op不为"d" 则添加状态中的数据
                broadcastState.put(sourceTable, value);
                //tableProcessHashMap.put(sourceTable, value);
            }
        } else {
            //DWD
            //获取op字段
            String op = value.getOp();
            //获取source_table字段
            String sourceTable = value.getSourceTable();
            String sourceType = value.getSourceType();
            String key = sourceTable + "-" + sourceType;
            //如果sink_type为dim,Op为"d"   则删除状态中的数据
            if ("d".equals(op)) {
                broadcastState.remove(key);
                tableProcessHashMap.remove(key);
            } else {
                //如果sink_type为dim,Op不为"d" 则添加状态中的数据
                broadcastState.put(key, value);
            }
        }
    }

    /**
     * 列过滤方法
     *
     * @param data        待过滤数据  {"id":"12","tm_name":"atguigu","logo_url":"/aaa/bbb"}
     * @param sinkColumns 过滤条件    id,tm_name
     */
    private void filterColumns(JSONObject data, String sinkColumns) {

        String[] split = sinkColumns.split(",");
        List<String> columnsList = Arrays.asList(split);

//        Set<Map.Entry<String, Object>> entries = data.entrySet();
//        Iterator<Map.Entry<String, Object>> iterator = entries.iterator();
//        while (iterator.hasNext()) {
//            Map.Entry<String, Object> next = iterator.next();
//            if (!columnsList.contains(next.getKey())) {
//                iterator.remove();
//            }
//        }

        Set<Map.Entry<String, Object>> entries = data.entrySet();
        entries.removeIf(next -> !columnsList.contains(next.getKey()));
    }
}
